using System.Collections.Concurrent;
using Shielded;
using Shielded.Gossip;

namespace ConsoleApp1;



public class Topic<T>
{
    public delegate void TopicMessageHandler<T>(T msg);
    private readonly GossipBackend _backend;
    // topic -> TopicQueue
    private readonly ConcurrentDictionary<string, ConcurrentQueue<object>> _topicQueues = new();
    // topic -> 多播委托
    private readonly ConcurrentDictionary<string, Delegate> _delegateMap = new();
    
    private readonly AutoResetEvent _resetEvent = new (false);

    private string topicName;
    
    // 静态计数器，用于生成唯一的key和版本号
    private static int _counter = 0;
    private static int NextMessageId(){
        if (_counter > int.MaxValue - 1000) // 接近最大值时重置
        {
            Interlocked.Exchange(ref _counter, 1); // 重置计数器
        }
        return Interlocked.Increment(ref _counter);
    }

    public Topic(GossipBackend backend,string topicName)
    {
        this.topicName = topicName;
        this._backend = backend;
        Shield.InTransaction(() => _backend.Changed.Subscribe(OnBackendChanged));
        //初始化队列
        var topicQueue = _topicQueues.GetOrAdd(topicName, _ => new ConcurrentQueue<object>());
        Task.Run(() =>
        {
            while (true)
            {
                _resetEvent.WaitOne(); // 阻塞直到有新消息
                while (topicQueue.TryDequeue(out var msg))
                {
                    if (_delegateMap.TryGetValue(topicName, out var del) && del is TopicMessageHandler<T> multiHandler)
                    {
                        // 类型不对会抛异常
                        try { multiHandler((T)msg); }
                        catch (Exception ex) { Console.WriteLine($"订阅者处理消息时发生错误: {ex.Message}"); }
                    }
                }
            }
        });
    }

    private void OnBackendChanged(object? sender, ChangedEventArgs e)
    {
        if (e.Deleted || e.NewValue == null)
            return;
        // 只处理本topic的消息
        if (!e.Key.StartsWith($"topic:{topicName}:")) return;
        var topicQueue = _topicQueues.GetOrAdd(topicName, _ => new ConcurrentQueue<object>());
        if (e.NewValue is Multiple<TopicEnvelope<T>> multiple)
        {
            foreach (var envelope in multiple)
            {
                if (envelope?.Messages != null && envelope.Messages.Count > 0)
                {
                    foreach (var msg in envelope.Messages)
                    {
                        topicQueue.Enqueue(msg);
                        _resetEvent.Set();
                    }
                    // 消费后移除该消息key
                    //Shield.InTransaction(() => _backend.Remove(e.Key));
                }
            }
        }
        else if (e.NewValue is TopicEnvelope<T> envelope)
        {
            Console.WriteLine("TopicEnvelope Message not handle");
        }
    }

    /// <summary>
    /// 发布消息到话题（支持任意类型）
    /// </summary>
    public void Publish(T message)
    {
        // 使用递增计数器生成唯一key
        var messageId = NextMessageId();
        var key = $"topic:{topicName}:{_backend.Transport.OwnId}:{messageId}";
        var envelope = new TopicEnvelope<T>
        {
            Messages = new List<T> { message },
            Version = new VersionVector(_backend.Transport.OwnId, messageId)
        };
        
        // 设置有效期，防止消息滞留
        Shield.InTransaction(() => _backend.SetHasVec(key, envelope, expireInMs: 20000));
    }

    /// <summary>
    /// 订阅话题（AutoResetEvent事件唤醒，无消息时等待）
    /// </summary>
    public void Subscribe(TopicMessageHandler<T> handler)
    {
        _delegateMap.AddOrUpdate(topicName, handler,
            (k, existing) => Delegate.Combine(existing, handler));

    }

    /// <summary>
    /// 取消订阅
    /// </summary>
    public void Unsubscribe(TopicMessageHandler<T> handler)
    {
        _delegateMap.AddOrUpdate(topicName,
            null,
            (k, existing) => Delegate.Remove(existing, handler));
    }
}


/// <summary>
/// 用于封装消息列表并实现 IHasVersionVector
/// </summary>
public class TopicEnvelope<T> : IHasVersionVector
{
    public List<T> Messages { get; set; } = new List<T>();
    public VersionVector Version { get; set; } = new VersionVector();
}